package com.ocean.streamingrecommend

import java.sql.{Connection, PreparedStatement, ResultSet}

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}


/**
  * mysql连接配置
  *
  * @param uri
  * @param user
  * @param password
  */
case class MysqlConfig(uri: String, user: String, password: String)

/**
  * 电影相似度列表
  *
  * @param mid
  * @param sim_mid
  * @param sim_score
  */
case class MovieRecommendation(mid: Int, sim_mid: Int, sim_score: Double)

/**
  * 实时推荐
  */
object StreamingRecommend {

  //定义常量
  //用户最近的K次评分
  val MAX_USER_RATINGS_NUM = 20
  //备选电影的个数
  val MAX_SIM_MOVIES_NUM = 20
  //每个用户实时推荐的数量
  val STREAMING_RECOMMENDATION_NUM_FOR_A_USER = 5
  //备选电影和最近评分电影相似度的阈值
  val CANDIDATE_RECENTLY_RATING_MOVIE_SIM_SCORE_THRESHOLD = 0.8
  //高评分阈值
  val HIGH_RATING_THRESHOLD = 3
  //流式推荐表
  val STREAMING_RECOMMENDATION = "streaming_recommendation"
  //评分表
  val MySql_RATING_Table = "rating"
  //电影相似度矩阵表
  val MOVIE_RECOMMENDATION = "movie_recommendation"


  def main(args: Array[String]): Unit = {
    val config = Map(
      "spark.cores" -> "local[*]",
      "mysql.uri" -> "jdbc:mysql://192.168.10.105:3306/recommend?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true&useSSL=false",
      "mysql.user" -> "root",
      "mysql.password" -> "cde32wsxzaq1",
      "kafka.topic" -> "recommend"
    )
    // 创建一个sparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("StreamingRecommend").set("spark.driver.maxResultSize", "10g")

    // 创建一个SparkSession
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    val sc: SparkContext = spark.sparkContext

    sc.setLogLevel("WARN")

    val streamingContext = new StreamingContext(sc, Seconds(2))

    import spark.implicits._

    implicit val mysqlConfig = MysqlConfig(config("mysql.uri"), config("mysql.user"), config("mysql.password"))

    val smMovieMatrix: collection.Map[Int, Map[Int, Double]] = spark.read.format("jdbc")
      .option("url", mysqlConfig.uri)
      .option("user", mysqlConfig.user)
      .option("password", mysqlConfig.password)
      .option("dbtable", MOVIE_RECOMMENDATION)
      .load()
      .as[MovieRecommendation].rdd
      .map(data => (data.mid, (data.sim_mid, data.sim_score)))
      .groupByKey()
      .map {
        case (mid, it) => (mid, it.map(x => (x._1, x._2)).toMap)
      }.collectAsMap()

    println("start streaming-recommend.............")
    val simMovieBroadcast: Broadcast[collection.Map[Int, Map[Int, Double]]] = sc.broadcast(smMovieMatrix)

    // 定义kafka连接参数

    val kafkaConfMap: Map[String, Object] = Map[String, Object](ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092, hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "recommend",
      "key.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "auto.offset.reset" -> "latest")

    val record: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set(config("kafka.topic")), kafkaConfMap))

    //处理原始的评分数据(原始评分数据格式：uid|mid|score|timestamp)
    val ratingStream: DStream[(Int, Int, Double, Long)] = record.map {
      msg => {
        val attr: Array[String] = msg.value().split("\\|")
        (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toLong)
      }
    }

    //实时推荐核心算法
    ratingStream.foreachRDD {
      rdds =>
        rdds.foreach {
          case (uid, mid, score, timestamp) => {
            println("data is coming from kafka stream>>>>>>>>>>>>>>" + uid + "|" + mid + "|" + score + "|" + timestamp)

            //从redis中获取当前用户的最近K此评分
            val userRecentlyRatings: Array[(Int, Double)] = getUserRecentlyRating(MAX_USER_RATINGS_NUM, uid)

            //从相似度矩阵中取出和当前电影最相似的K个电影作为备选
            val candidateMovies: Array[Int] = getTopSimMovies(MAX_SIM_MOVIES_NUM, mid, uid, simMovieBroadcast.value)

            //对于每个备选电影，计算推荐优先级，得到当前用户的实时推荐列表
            val streamRecommendationMovies: Array[(Int, Double)] = computeRecommendLevel(candidateMovies, userRecentlyRatings, simMovieBroadcast.value)

            //把数据保存到mysql展现到前端
            saveDataToMySql(uid, streamRecommendationMovies)
            println("data is saved in mysql>>>>>>>>>>>>>>" + uid + "|" + mid + "|" + score + "|" + timestamp)

          }
        }
    }

    streamingContext.start()
    streamingContext.awaitTermination()
  }

  import scala.collection.JavaConversions._

  /**
    * 从redis获取某个用户最近K次的电影评分数据
    *
    * @param MAX_USER_RATINGS_NUM
    * @param uid
    * @return
    */
  def getUserRecentlyRating(MAX_USER_RATINGS_NUM: Int, uid: Int): Array[(Int, Double)] = {
    val jedis = new Jedis("192.168.10.105")

    //redis中的数据格式是key-->uid:1  value-->queue(mid:score)
    jedis.lrange("uid:" + uid, 0, MAX_USER_RATINGS_NUM - 1)
      .map {
        item => {
          //得到一组电影ID:电影评分
          val attr: Array[String] = item.split("\\:")
          (attr(0).trim.toInt, attr(1).trim.toDouble)
        }
      }.toArray
  }

  /**
    * 获取相似的电影并且取出已经看过的
    *
    * @param MAX_SIM_MOVIES_NUM
    * @param mid
    * @param uid
    * @param simMovies
    * @param mysqlConfig
    * @return
    */
  def getTopSimMovies(MAX_SIM_MOVIES_NUM: Int, mid: Int, uid: Int, simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])(implicit mysqlConfig: MysqlConfig): Array[Int] = {
    //获取所有相似的电影
    val allSimMovies: Array[(Int, Double)] = simMovies(mid).toArray

    //从mysql中得到用户已经看过的电影，因为已经看过的电影我们就不推了
    val connection: Connection = JDBCUtil.getConnection

    val allAlreadySeenMoviesSql =
      s"""
         |select mid from $MySql_RATING_Table where uid=?
      """.stripMargin
    val ps: PreparedStatement = connection.prepareStatement(allAlreadySeenMoviesSql)
    ps.setInt(1, uid)
    val rs: ResultSet = ps.executeQuery()
    val midList = ListBuffer[Int]()
    while (rs.next()) {
      val mid: Int = rs.getInt("mid")
      midList += mid
    }

    //把已经看过的电影过滤掉
    allSimMovies.filter(x => !midList.contains(x._1))
      .sortWith(_._2 > _._2)
      .take(MAX_SIM_MOVIES_NUM)
      .map(x => x._1)
  }


  /**
    * 根据实时推荐设计的公式计算每部备选电影的推荐等级
    *
    * @param candidateMovies
    * @param userRecentlyRatings
    * @param simMovies
    */
  def computeRecommendLevel(candidateMovies: Array[Int], userRecentlyRatings: Array[(Int, Double)], simMovies: collection.Map[Int, Map[Int, Double]]) = {
    //每个电影的基础评分
    val scores: ArrayBuffer[(Int, Double)] = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
    //增加因子
    val enhanceFactor: mutable.HashMap[Int, Int] = scala.collection.mutable.HashMap[Int, Int]()
    //减弱因子
    val weakenFactor: mutable.HashMap[Int, Int] = scala.collection.mutable.HashMap[Int, Int]()

    //对备选电影和我最近评分电影做一次双重循环
    for (candidateMovie <- candidateMovies; userRecentlyRating <- userRecentlyRatings) {
      //获取备选电影和最近评分电影的相似度
      val simScore: Double = getMovieSimScore(candidateMovie, userRecentlyRating._1, simMovies)
      //如果相似都大于某个阈值
      if (simScore > CANDIDATE_RECENTLY_RATING_MOVIE_SIM_SCORE_THRESHOLD) {
        //计算备选电影的基础得分
        scores += ((candidateMovie, simScore * userRecentlyRating._2))
        //计算偏移量
        if (userRecentlyRating._2 > HIGH_RATING_THRESHOLD) {
          //增强因子
          enhanceFactor(candidateMovie) = enhanceFactor.getOrDefault(candidateMovie, 0) + 1
        } else {
          //减弱因子
          weakenFactor(candidateMovie) = weakenFactor.getOrDefault(candidateMovie, 0) + 1
        }
      }
    }
    //完成循环之后，我们有了每一部备选电影基础得分以及它对应的增强因子和减弱因子
    //因为一部备选电影可能和我最近看过的几部电影的相似度都有超过0.7，所以这里需要用group by
    scores.groupBy(_._1).map {
      //这里的scoreList是一个ArrayBuffer[(mid, score)]数据结构
      case (mid, scoreList) => {
        (mid, scoreList.map(_._2).sum / scoreList.length + log(enhanceFactor.getOrDefault(mid, 1)) - log(weakenFactor.getOrDefault(mid, 1)))
      }
    }.toArray.sortWith(_._2 > _._2).take(STREAMING_RECOMMENDATION_NUM_FOR_A_USER)
  }

  /**
    * 这里单独作为一个方法是可以作为参数（对数的底数）调节
    *
    * @param m
    * @return
    */
  def log(m: Int) = {
    val N = 10
    math.log(m) / math.log(N)
  }

  /**
    *
    * @param mid1      备选电影
    * @param mid2      最近K次评分的电影
    * @param simMovies 所有的电影相似度
    * @return 备选电影mid1和我最近K次评分中的某个电影mid2的相似度
    */
  def getMovieSimScore(mid1: Int, mid2: Int, simMovies: scala.collection.Map[Int, Map[Int, Double]]) = {
    simMovies.get(mid1) match {
      case Some(sims) => sims.get(mid2) match {
        case Some(score) => score
        case None => 0.0
      }
      case None => 0.0
    }
  }

  /** 将给用户uid的实施推荐数据保存到mysql
    *
    * @param uid
    * @param streamRecommendationMovies
    * @param mysqlConfig
    */
  def saveDataToMySql(uid: Int, streamRecommendationMovies: Array[(Int, Double)])(implicit mysqlConfig: MysqlConfig): Unit = {
    val connection: Connection = JDBCUtil.getConnection

    val createSql =
      s"""
         |create table if not exists $STREAMING_RECOMMENDATION
         |(
         |uid int not null,
         |mid int not null,
         |recommend_level double not null,
         |key idx_uid (uid),
         |key idx_mid (mid)
         |)engine=InnoDB default charset=utf8mb4
        """.stripMargin

    connection.prepareStatement(createSql).execute()

    val findAndRemoveSql =
      s"""
         |delete from  $STREAMING_RECOMMENDATION
         |where uid=?
      """.stripMargin
    var ps: PreparedStatement = connection.prepareStatement(findAndRemoveSql)
    ps.setInt(1, uid)
    ps.executeUpdate()

    val insertSql =
      s"""
         |insert into $STREAMING_RECOMMENDATION(uid, mid, recommend_level)
         values(?,?,?)
      """.stripMargin
    ps = connection.prepareStatement(insertSql)
    for (elem <- streamRecommendationMovies) {
      ps.setInt(1, uid)
      ps.setInt(2, elem._1)
      ps.setDouble(3, elem._2)
      ps.executeUpdate()
    }

    ps.close()
    connection.close()
  }


}

